package com.jw.debezuim.config.debezium;

import io.debezium.config.Configuration;
import io.debezium.embedded.EmbeddedEngine;
import io.debezium.engine.DebeziumEngine;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
@Order(Integer.MAX_VALUE - 1)
public class DebezuimConfig implements ApplicationRunner {

    private EmbeddedEngine engine;

    @Value("${test.col:'a','b','c'}")
    private Set<String> testCol;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("o=|>> 开始配置debezuim <<|=o");
        // Define the configuration for the embedded and MySQL connector ...
        Configuration config = Configuration.create()
                /* begin engine properties */
                .with("name", "zt3000-debezuim-mysql-connector")
                .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", "./storage/offset.dat")
                .with("offset.flush.interval.ms", 60000)
                .with("snapshot.mode", "when_needed") //initial/when_needed/schema_only/schema_only_recovery/never
                .with("snapshot.locking.mode", "minimal") //minimal/minimal_percona/extended/none
//                .with("snapshot.include.collection.list", "All tables specified in table.include.list")
                .with("signal.data.collection", "debezuim.debezuim_signal")
                /* begin connector properties
                * https://debezium.io/documentation/reference/2.1/connectors/mysql.html#mysql-property-column-propagate-source-type
                */
                .with("database.hostname", "192.168.17.238")
                .with("database.port", 3306)
                .with("database.user", "root")
                .with("database.password", "root")
                .with("database.server.id", 85744)
                .with("topic.prefix", "zt3000-debezuim-mysql-connector")
                .with("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory")
                .with("schema.history.internal.file.filename", "./storage/schemahistory.dat")
                .with("database.include.list", "debezuim,jw_test,test_binlog")
                .with("table.include.list", "jw_test.product_view,test_binlog.tb_metadata_plat_cat")
                .with("include.schema.changes", true)
                .with("include.query", false) // binlog_rows_query_log_events option set to ON
                .with("max.batch.size", 1024)
                .with("max.queue.size", 8192)
                .with("poll.interval.ms", 100)
                .with("connect.timeout.ms", 30000)
//                .with("message.key.columns", "test_binlog.tb_metadata_plat_cat:name,type") //inventory.customers:pk1,pk2;(.*).purchaseorders:pk3,pk4
                .with("snapshot.fetch.size", 1024) //During a snapshot, the connector reads table content in batches of rows. This property specifies the maximum number of rows in a batch.

                .build();
        // Create the engine with this configuration ...
        DebeziumEngine.CompletionCallback completionCallback = new DebeziumEngine.CompletionCallback() {
            @Override
            public void handle(boolean success, String message, Throwable error) {
                log.info("debezuim运行结束,状态=>{},消息=>{}", success, message);
                if (error != null) {
                    log.error("debezuim运行异常结束", error);
                }
            }
        };
        this.engine = (EmbeddedEngine) EmbeddedEngine.create()
                .using(config)
                .using(completionCallback)
                .notifying(new EventChangeConsumer())
                .build();

        // Run the engine asynchronously ...
        Executor executor = Executors.newSingleThreadExecutor((runnable) -> new Thread(runnable, "debezuim-engine-executor"));
        executor.execute(engine);
    }

    private void handleEvent(SourceRecord sourceRecord) {
        log.info("SourceRecord=> {}", sourceRecord);
        Object value = sourceRecord.value();
    }

    @PreDestroy
    public void preDestroy() {
        log.info("o=|>> 开始关闭debezuim <<|=o");
        if (this.engine != null) {
            try {
                this.engine.close();
            } catch (IOException ignore) {}
            try {
                while (!engine.await(5, TimeUnit.SECONDS)) {
                    log.info("Wating another 5 seconds for the embedded engine to shut down");
                }
            }
            catch ( InterruptedException ignore ) {}
            log.info("o=|>> 关闭debezuim完成 <<|=o");
        }
    }

    public static void main(String[] args) {
    }
}
